"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 数据库系统
Case Name   : 打开快速负载均衡，建连完成后，修改建连用户密码后等待后台线程检测完成
Create At   : 2023/04/12
Owner       : @peilinqian
Description :
    step0: 检查集群状态，不正常时重启;
    step1: 设置用户密码修改认证相关参数;
    step2: 创建指定用户;
    step3: 创建指定数据库;
    step4: 远程复制java脚本至指定主机;
    step5: 编译java文件;
    step6: 以线程运行java脚本;
    step7: 等待2s，查询节点连接数量;
    step8: 修改用户密码;
    step9: java脚本执行22s左右时，超过后台线程检测时间点，查询三个节点连接数量;
    step10: 恢复用户密码;
    step11: java脚本执行32s左右时，新建15个连接后，查询三个节点连接数量;
    step12：验证step5的Java执行结果日志;
Expect      :
    step0: 检查集群状态，不正常时重启; expect:状态正常
    step1: 设置用户密码修改认证相关参数; expect:修改成功
    step2: 创建指定用户; expect:创建成功
    step3: 创建指定数据库; expect:创建成功
    step4: 远程复制java脚本至指定主机; expect:复制成功
    step5: 编译java文件; expect:编译成功
    step6: 以线程运行java脚本; expect: 操作成功
    step7: 等待2s，查询节点连接数量; expect: 查询结果均为4
    step8: 修改用户密码; expect: 修改成功
    step9: java脚本执行22s左右时，超过后台线程检测时间点，查询三个节点连接数量; expect: 关闭失效连接，查询结果均为0
    step10: 恢复用户密码; expect: 修改成功
    step11: java脚本执行32s左右时，新建15个连接后，查询三个节点连接数量; expect: 查询结果均为5
    step12：验证step5的Java执行结果日志; expect:包含节点探活及缓存线程、关闭连接线程、及关键日志
History     :
"""
import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

pri_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(pri_sh.get_node_num() < 3, '非1主2备环境不执行')
class JDBCQuikAutoBalance(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info("----this is setup----")
        self.log.info(f'----{os.path.basename(__file__)}:start----')
        self.pri_root = Node(node='PrimaryRoot')
        self.sta1_sh = CommonSH('Standby1DbUser')
        self.sta2_sh = CommonSH('Standby2DbUser')
        self.db_name = 'db_jdbc_quickautobalance_case0022'
        self.u_name = 'u_jdbc_quickautobalance_case0022'
        self.target_path = os.path.join(macro.DB_BACKUP_PATH, 'JDBC_test')
        self.java_name = "JDBC_QuikAutoBalance_Case0022"
        self.com = Common()
        self.constant = Constant()
        self.assert_flag1 = 'Start scheduleExecutorService'
        self.assert_flag2 = 'Start closeConnectionScheduledFuture'
        self.assert_flag3 = 'A data node failed, clear cached connections'

    def test_main(self):
        step_txt = '----step0: 检查集群状态，不正常时重启; expect:状态正常----'
        self.log.info(step_txt)
        if not pri_sh.get_db_cluster_status('status'):
            self.assertTrue(pri_sh.restart_db_cluster(), '执行失败' + step_txt)
        step_txt = '----查询用户密码修改认证相关参数初始配置值----'
        self.log.info(step_txt)
        self.init_para1 = self.com.show_param("password_reuse_time")
        self.init_para2 = self.com.show_param("password_reuse_max")
        self.init_para3 = self.com.show_param("failed_login_attempts")

        step_txt = '----step1: 设置用户密码修改认证相关参数; expect:修改成功----'
        self.log.info(step_txt)
        result = pri_sh.execute_gsguc('reload',
                                      self.constant.GSGUC_SUCCESS_MSG,
                                      f"password_reuse_time= 0")
        self.assertTrue(result, '执行失败:' + step_txt)
        result = pri_sh.execute_gsguc('reload',
                                      self.constant.GSGUC_SUCCESS_MSG,
                                      f"password_reuse_max= 0")
        self.assertTrue(result, '执行失败:' + step_txt)
        result = pri_sh.execute_gsguc('reload',
                                      self.constant.GSGUC_SUCCESS_MSG,
                                      f"failed_login_attempts= 0")
        self.assertTrue(result, '执行失败:' + step_txt)

        step_txt = '----step2: 创建指定用户; expect:创建成功----'
        self.log.info(step_txt)
        create_sql = f"drop user if exists {self.u_name};" \
            f"create user {self.u_name} with " \
            f"password '{pri_sh.node.db_password}';" \
            f"grant all privileges to {self.u_name};"
        create_result = pri_sh.execut_db_sql(create_sql)
        self.log.info(create_result)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, create_result,
                      '执行失败' + step_txt)

        step_txt = '----step3: 创建指定数据库; expect:创建成功----'
        self.log.info(step_txt)
        create_sql = f"drop database if exists {self.db_name};" \
            f"create database {self.db_name};checkpoint;"
        create_result = pri_sh.execut_db_sql(create_sql)
        self.log.info(create_result)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, create_result,
                      '执行失败' + step_txt)
        time.sleep(2)

        step_txt = '----step4: 远程复制java脚本至指定主机; expect:复制成功----'
        self.log.info(step_txt)
        self.com.scp_file(self.pri_root, f'{self.java_name}.java',
                          self.target_path)
        check_cmd = f'ls {self.target_path}'
        check_result = self.pri_root.sh(check_cmd).result()
        self.assertIn(self.java_name, check_result, '执行失败' + step_txt)

        step_txt = '----step5: 编译java文件; expect:编译成功----'
        self.log.info(step_txt)
        self.pri_root.scp_put(macro.JDBC_PATH, os.path.join(self.target_path,
                                                            'postgresql.jar'))
        javac_cmd = f"javac -encoding utf-8 -cp " \
            f"{os.path.join(self.target_path, 'postgresql.jar')} " \
            f"{os.path.join(self.target_path, f'{self.java_name}.java')}"
        self.log.info(javac_cmd)
        javac_result = self.pri_root.sh(javac_cmd).result()
        self.log.info(javac_result)
        self.assertEqual('', javac_result, "执行失败" + step_txt)

        step_txt = '----step6: 以线程运行java脚本; expect: 操作成功----'
        self.log.info(step_txt)
        session1 = None
        try:
            url_host = f"{pri_sh.node.db_host}:{pri_sh.node.db_port}," \
                f"{self.sta1_sh.node.db_host}:{self.sta1_sh.node.db_port}," \
                f"{self.sta2_sh.node.db_host}:{self.sta2_sh.node.db_port}"
            url_param = "loggerLevel=info\&autoBalance=leastconn" \
                        "\&enableQuickAutoBalance=true"
            java_cmd = f"java -cp " \
                f"{os.path.join(self.target_path, 'postgresql.jar')}" \
                f":{self.target_path} {self.java_name} {url_host} " \
                f"{self.db_name} {self.u_name} " \
                f"{pri_sh.node.db_password} {url_param}"
            self.log.info(java_cmd)
            session1 = ComThread(self.com.get_sh_result,
                                 args=(self.pri_root, java_cmd,))
            session1.setDaemon(True)
            session1.start()
            start_time = time.time()

            step_txt = '----step7: 等待2s，查询节点连接数量; expect: 查询结果均为4----'
            self.log.info(step_txt)
            time.sleep(2)
            check_sql = f"select * from pg_stat_activity where " \
                f"application_name = 'PostgreSQL JDBC Driver' " \
                f"and datname='{self.db_name}';"
            self.log.info('----主节点连接数量----')
            pri_result = pri_sh.execut_db_sql(check_sql)
            self.log.info(pri_result)
            self.log.info('----备1节点连接数量----')
            sta1_result = self.sta1_sh.execut_db_sql(check_sql)
            self.log.info(sta1_result)
            self.log.info('----备2节点连接数量----')
            sta2_result = self.sta2_sh.execut_db_sql(check_sql)
            self.log.info(sta2_result)
            self.assertIn('4 rows', pri_result, "执行失败" + step_txt)
            self.assertIn('4 rows', sta1_result, "执行失败" + step_txt)
            self.assertIn('4 rows', sta2_result, "执行失败" + step_txt)

            step_txt = '----step8: 修改用户密码; expect: 修改成功----'
            self.log.info(step_txt)
            alter_sql = f"alter user {self.u_name} password 'Test@123';"
            alter_result = pri_sh.execut_db_sql(alter_sql)
            self.log.info(alter_result)
            self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, alter_result,
                          '执行失败' + step_txt)

            step_txt = '----step9: java脚本执行22s左右时，超过后台线程检测时间点，查询三个节点连接数量; ' \
                       'expect: 关闭失效连接，查询结果均为0----'
            self.log.info(step_txt)
            end_time = time.time()
            if end_time - start_time < 22:
                self.log.info('--不超过22s--')
                time.sleep(22 - end_time + start_time)
            else:
                self.log.info('--超过22s--')
            self.log.info('----主节点连接数量----')
            pri_result = pri_sh.execut_db_sql(check_sql)
            self.log.info(pri_result)
            self.log.info('----备1节点连接数量----')
            sta1_result = self.sta1_sh.execut_db_sql(check_sql)
            self.log.info(sta1_result)
            self.log.info('----备2节点连接数量----')
            sta2_result = self.sta2_sh.execut_db_sql(check_sql)
            self.log.info(sta2_result)
            self.assertIn('0 rows', pri_result, "执行失败" + step_txt)
            self.assertIn('0 rows', sta1_result, "执行失败" + step_txt)
            self.assertIn('0 rows', sta2_result, "执行失败" + step_txt)

            step_txt = '----step10: 恢复用户密码; expect: 修改成功----'
            self.log.info(step_txt)
            alter_sql = f"alter user {self.u_name} " \
                f"password '{pri_sh.node.db_password}';"
            alter_result = pri_sh.execut_db_sql(alter_sql)
            self.log.info(alter_result)
            self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, alter_result,
                          '执行失败' + step_txt)

            step_txt = '----step11: java脚本执行32s左右时，新建15个连接后，查询三个节点连接数量;' \
                       'expect: 查询结果均为5----'
            self.log.info(step_txt)
            end_time = time.time()
            if end_time - start_time < 32:
                self.log.info('--不超过32s--')
                time.sleep(32 - end_time + start_time)
            else:
                self.log.info('--超过32s--')
            self.log.info('----主节点连接数量----')
            pri_result = pri_sh.execut_db_sql(check_sql)
            self.log.info(pri_result)
            self.log.info('----备1节点连接数量----')
            sta1_result = self.sta1_sh.execut_db_sql(check_sql)
            self.log.info(sta1_result)
            self.log.info('----备2节点连接数量----')
            sta2_result = self.sta2_sh.execut_db_sql(check_sql)
            self.log.info(sta2_result)
            self.assertIn('5 rows', pri_result, "执行失败" + step_txt)
            self.assertIn('5 rows', sta1_result, "执行失败" + step_txt)
            self.assertIn('5 rows', sta2_result, "执行失败" + step_txt)
        finally:
            if session1:
                step_txt = '----step12：验证step5的Java执行结果日志; ' \
                           'expect:包含节点探活及缓存线程、关闭连接线程、及关键日志----'
                self.log.info(step_txt)
                session1.join()
                session1_result = session1.get_result()
                self.assertIn(self.assert_flag1, session1_result,
                              "执行失败" + step_txt)
                self.assertIn(self.assert_flag2, session1_result,
                              "执行失败" + step_txt)
                self.assertIn(self.assert_flag3, session1_result,
                              "执行失败" + step_txt)

    def tearDown(self):
        self.log.info('----this is teardown----')
        self.log.info('----清理环境----')
        ps_cmd = f"ps ux |grep {self.java_name}|grep -v grep|" \
            f"tr -s ' '|cut -d ' ' -f 2"
        self.log.info(ps_cmd)
        ps_result = self.pri_root.sh(ps_cmd).result()
        self.log.info(ps_result)
        if ps_result != '':
            self.log.info('----清理java进程----')
            kill_cmd = f"ps ux |grep {self.java_name}|grep -v grep|" \
                f"tr -s ' '|cut -d ' ' -f 2|xargs kill -9"
            self.log.info(kill_cmd)
            self.pri_root.sh(kill_cmd)

        step_txt1 = '----恢复用户密码修改认证相关参数; expect:修改成功----'
        self.log.info(step_txt1)
        guc_result1 = pri_sh.execute_gsguc('reload',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           f"password_reuse_time= "
                                           f"{self.init_para1}")

        guc_result2 = pri_sh.execute_gsguc('reload',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           f"password_reuse_max= "
                                           f"{self.init_para2}")

        guc_result3 = pri_sh.execute_gsguc('reload',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           f"failed_login_attempts= "
                                           f"{self.init_para3}")

        step_txt2 = '----删除用户; expect:删除成功----'
        self.log.info(step_txt2)
        drop_user_sql = f"drop user if exists {self.u_name};"
        drop_user_result = pri_sh.execut_db_sql(drop_user_sql)
        self.log.info(drop_user_result)

        step_txt3 = '----删除数据库; expect:删除成功----'
        self.log.info(step_txt3)
        drop_sql = f"drop database if exists {self.db_name};"
        drop_result = pri_sh.execut_db_sql(drop_sql)
        self.log.info(drop_result)

        step_txt4 = '----清理编译脚本; expect:清理成功----'
        self.log.info(step_txt4)
        file_rm_cmd = f'rm -rf {self.target_path};' \
            f'if [ -d {self.target_path} ]; ' \
            f'then echo "exists"; else echo "not exists"; fi'
        self.log.info(file_rm_cmd)
        file_rm_result = self.pri_root.sh(file_rm_cmd).result()
        self.log.info(file_rm_result)

        self.log.info('----清理环境断言----')
        self.assertTrue(guc_result1, '执行失败:' + step_txt1)
        self.assertTrue(guc_result2, '执行失败:' + step_txt1)
        self.assertTrue(guc_result3, '执行失败:' + step_txt1)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, drop_user_result,
                      '执行失败' + step_txt2)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, drop_result,
                      '执行失败' + step_txt3)
        self.assertEqual('not exists', file_rm_result, "执行失败" + step_txt4)

        self.log.info(f'----{os.path.basename(__file__)}:end----')
